package com.example.kafka.service;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Properties;

/**
 * 反例：批量发送时每条都新建 Producer 实例
 */
@Service
public class ProducerServiceBatchSendWrong {

    public void sendBatch(String topic, List<String> messages) {
        for (String message : messages) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            KafkaProducer<String, String> producer = new KafkaProducer<>(props);
            try {
                producer.send(new ProducerRecord<>(topic, message));
            } finally {
                producer.close();
            }
        }
    }
}